package com.hsc.www.amqp.config;//package com.hsc.www.amqp.config;

import com.hsc.www.amqp.util.RabbitMetaMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


/**
 * <p><b>Description:</b> RabbitTemplate配置工厂类
 * <p><b>Company:</b>
 *
 * @author created by hongda at 11:33 on 2017-07-05
 * @version V0.1
 */
@Configuration
public class Config {
    private Logger logger = LoggerFactory.getLogger(Config.class);

    @Autowired
    MessageTransientStore messageTransientStore;

    @Autowired
    MessageDurableStore messageDurableStore;

    @Autowired
    RedisTemplate redisTemplate;

    ConcurrentMap<String, String> returnCallbackMessage = new ConcurrentHashMap<>();


    /**
     * 对RabbitStarter创建的rabbitTemplate进行配置。
     *
     * @param rabbitTemplate
     */
    @Autowired
    public void rabbitTemplate(RabbitTemplate rabbitTemplate) {

        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        /**
         * ReturnCallback和ConfirmCallback的讲解
         *
         *       配置文件:
         *          spring:
         *              rabbitmq:
         *                  publisher-returns: true
         *                  publisher-confirms: true
         *
         *        如果消息没有到exchange,则confirmCallback回调,ack=false
         *        如果消息到达exchange,则confirmCallback回调,ack=true
         *        如果消息到从exchange到queue成功,则不回调returnCallback
         *        如果消息到从exchange到queue失败,则回调returnCallback(需设置rabbitTemplate.setMandatory(true),否则不回回调,消息就丢了。如果没有手动设置将使用配文件中的publisher-returns）
         *
         *        有种极端情况：ack为false ，错误原因为clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)
         *        如果是这种情况消息其实已经存入队列中了。这种情况在测试的时候经常会出现，就是程序主线启动后，发送了mq消息后就马上结束了,并关闭了channel，留下子线程在监听mq向应，子线程收到消息后在关闭channel时出现异常，可以在
         *        主线程后加一个while(true) 或  Thread.sleep() 让主线程等一会。
         *
         *
         *        如果有returnCallback 先于confirmCallback执行。
         */


        /**
         * 消息发送到RabbitMQ服务器，但无相应Exchange时的回调。
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            String messageId = message.getMessageProperties().getMessageId();
            returnCallbackMessage.put(messageId, "replyCode=" + replyCode + ";replyText=" + replyText);
        });

        /**
         * 消息发送到RabbitMQ交换器后接收ack回调https://www.jianshu.com/p/2c5eebfd0e95
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            boolean sendResult = false;
            String errorMsg = cause;
            if (ack) {
                if (returnCallbackMessage.containsKey(correlationData.getId())) {
                    errorMsg = returnCallbackMessage.get(correlationData.getId());
                    logger.error("returnCallback发生:" + errorMsg);
                    returnCallbackMessage.remove(correlationData.getId());
                } else {
                    sendResult = true;
                    logger.info("消息已正确投递到队列");
                }
            } else {//ack为false时不代表消息一定没有存入mq队列。
                logger.error("消息投递失败 或 消息投递结果未成功获取 ： " + cause);
            }
            store(((MyCorrelationData) correlationData).getMessage(), sendResult, errorMsg);
        });
    }


    public void store(RabbitMetaMessage message, boolean sendResult, String errorMsg) {
        messageDurableStore.saveMessage(message, sendResult, errorMsg);
        messageTransientStore.removeMessage(message.getMessageId());
    }


    @Bean
    public MessageDurableStore sdf() {

        return new MessageDurableStore() {
            @Override
            public void saveMessage(RabbitMetaMessage message, boolean sendResult, String errorMsg) {
                Map map = new HashMap();
                map.put("message", message);
                map.put("sendResult", sendResult);
                map.put("errorMsg", errorMsg);
                redisTemplate.opsForValue().set("qmq:durable:" + message.getExchange() + ":" + message.getRoutingKey() + ":" + message.getMessageId(), map);
            }
        };
    }

    @Bean
    public MessageTransientStore sdf1() {
        return new MessageTransientStore() {
            @Override
            public void saveMessage(RabbitMetaMessage message) {
                redisTemplate.opsForValue().set("qmq:transient:" + message.getMessageId(), message);
            }

            @Override
            public void removeMessage(String messageId) {
                redisTemplate.delete("qmq:transient:" + messageId);
            }

            @Override
            public RabbitMetaMessage getMessage(String messageId) {
                Object value = redisTemplate.opsForValue().get("qmq:transient:" + messageId);
                return (RabbitMetaMessage) value;
            }
        };
    }

    /**
     * 死信队列的监听
     *
     * @param   connectionFactory         RabbitMQ连接工厂
     * @param   deadLetterMessageListener 死信队列监听
     * @return  监听容器对象
     */
//    @Bean
//    public SimpleMessageListenerContainer deadLetterListenerContainer(ConnectionFactory connectionFactory,
//                                                                      MyMessageListener deadLetterMessageListener) {
//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
//        container.setQueueNames("1");
//        container.setExposeListenerChannel(true);
//        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//        container.setMessageListener(deadLetterMessageListener);
//        /**设置消费者能处理消息的最大个数*/
//        container.setPrefetchCount(100);
//        return container;
//    }

//    @Bean
//     public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
//        Jackson2JsonMessageConverter jsonMessageConverter = new Jackson2JsonMessageConverter();
//        return jsonMessageConverter;
//     }


}
